package com.lefeng.cockpit.datax.thread;

import com.lefeng.cockpit.common.bean.po.datasource.JobRegistryPO;
import com.lefeng.cockpit.common.bean.po.executor.ExecutorPO;
import com.lefeng.cockpit.common.enums.ExecutorStatusEnum;
import com.lefeng.cockpit.datax.conf.JobAdminConfig;
import com.lefeng.cockpit.datax.core.enums.RegistryConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Monitors job registry and updates the status of executors.
 * This class periodically checks and refreshes the registry information
 * of job executors and updates their status based on their availability.
 */
public class JobRegistryMonitorHelper {
    private static final Logger logger = LoggerFactory.getLogger(JobRegistryMonitorHelper.class);
    private static final JobRegistryMonitorHelper instance = new JobRegistryMonitorHelper();
    private Thread registryThread;
    private volatile boolean stop = false;

    private JobRegistryMonitorHelper() {
    }

    /**
     * Gets the singleton instance of JobRegistryMonitorHelper.
     *
     * @return The singleton instance.
     */
    public static JobRegistryMonitorHelper getInstance() {
        return instance;
    }

    /**
     * Starts the monitoring thread which will periodically check and update job registry information.
     */
    public void start() {
        registryThread = new Thread(this::run);
        registryThread.setDaemon(true);
        registryThread.setName("JobRegistryMonitorHelper");
        registryThread.start();
    }

    /**
     * The main logic of the monitoring thread.
     * It performs the following tasks:
     * 1. Removes dead registry entries.
     * 2. Updates the status of executors based on the current registry information.
     */
    private void run() {
        while (!stop) {
            try {
                // Remove dead registry entries
                removeDeadRegistryEntries();

                // Update executor status
                updateExecutorStatus();

                // Sleep for the configured interval
                TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
            } catch (Exception e) {
                if (!stop) {
                    logger.error("Job registry monitor thread error: {}", e.getMessage(), e);
                }
            }
        }

        logger.info("Job registry monitor thread stopped");
    }

    /**
     * Removes dead registry entries from the database.
     */
    private void removeDeadRegistryEntries() {
        LocalDateTime now = LocalDateTime.now();
        List<Integer> deadRegistryIds = JobAdminConfig.getAdminConfig().getJobRegistryDao()
                .findDead(RegistryConfig.DEAD_TIMEOUT, Date.from(now.toInstant(ZoneOffset.UTC)));
        if (deadRegistryIds != null && !deadRegistryIds.isEmpty()) {
            JobAdminConfig.getAdminConfig().getJobRegistryDao().removeDead(deadRegistryIds);
        }
    }

    /**
     * Updates the status of executors based on the current registry information.
     */
    private void updateExecutorStatus() {
        List<ExecutorPO> executors = JobAdminConfig.getAdminConfig().getExecutorDao().findByAddressType(0);
        if (executors != null && !executors.isEmpty()) {
            Map<String, List<String>> appAddressMap = buildAppAddressMap();
            for (ExecutorPO executor : executors) {
                updateExecutor(executor, appAddressMap);
            }
        }
    }

    /**
     * Builds a map of application names to their corresponding list of addresses.
     *
     * @return A map of application names to addresses.
     */
    private Map<String, List<String>> buildAppAddressMap() {
        LocalDateTime now = LocalDateTime.now();
        Map<String, List<String>> appAddressMap = new HashMap<>();
        List<JobRegistryPO> registryEntries = JobAdminConfig.getAdminConfig().getJobRegistryDao()
                .findAll(RegistryConfig.DEAD_TIMEOUT, Date.from(now.toInstant(ZoneOffset.UTC)));
        if (registryEntries != null) {
            for (JobRegistryPO entry : registryEntries) {
                if (RegistryConfig.RegistType.EXECUTOR.name().equals(entry.getRegistryGroup())) {
                    String appName = entry.getRegistryKey();
                    appAddressMap.computeIfAbsent(appName, k -> new ArrayList<>())
                            .add(entry.getRegistryValue());
                }
            }
        }
        return appAddressMap;
    }

    /**
     * Updates the status and address list of an executor.
     *
     * @param executor      The executor to update.
     * @param appAddressMap The map of application names to addresses.
     */
    private void updateExecutor(ExecutorPO executor, Map<String, List<String>> appAddressMap) {
        List<String> addresses = appAddressMap.get(executor.getAppName());
        if (addresses != null && !addresses.isEmpty()) {
            Collections.sort(addresses);
            String addressListStr = String.join(",", addresses);
            executor.setAddressList(addressListStr);
            executor.setStatus(ExecutorStatusEnum.LIVE.getCode());
        } else {
            executor.setAddressList("");
            executor.setStatus(ExecutorStatusEnum.DOWN.getCode());
        }
        JobAdminConfig.getAdminConfig().getExecutorDao().update(executor);
    }

    /**
     * Stops the monitoring thread and waits for it to terminate.
     */
    public void toStop() {
        stop = true;
        registryThread.interrupt();
        try {
            registryThread.join();
        } catch (InterruptedException e) {
            logger.error("Thread interrupted: {}", e.getMessage(), e);
        }
    }
}
